package com.rem.flink.flink1Base;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * 有界流处理
 *
 * @author Rem
 * @date 2022-09-26
 */

public class BoundedStreamWordCount {
    public static void main(String[] args) throws Exception {
        //1.创建执行环境
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.从文件中读取数据 按行读取
        DataStreamSource<String> dataStreamSource = streamEnv.readTextFile("input/word.txt");
        //3.转换数据格式 到元组
        SingleOutputStreamOperator<Tuple2<String, Long>> stremWordAndOne = dataStreamSource.flatMap((String line, Collector<String> words) -> {
                    Arrays.stream(line.split(" ")).forEach(words::collect);
                })
                .returns(Types.STRING)
                .map(word -> Tuple2.of(word, 1L))
                .returns(Types.TUPLE(Types.STRING, Types.LONG));

        //4.按照word 进行分组 根据Tuple2里第一个字段
        KeyedStream<Tuple2<String, Long>, String> wordAndOneKeyStream = stremWordAndOne.keyBy(tuple2 -> tuple2.f0);
        //5.分组内聚合统计 根据Tuple2里第二个字段
        SingleOutputStreamOperator<Tuple2<String, Long>> sum = wordAndOneKeyStream.sum(1);
        //6.打印结果
        sum.print();

        //7.执行流
        streamEnv.execute();

    }
}
